#include "sys.h"
#include "net.h"
#include "log.h"
#include "thread_pool.h"
#include "platform.h"
#include "cJSON.h"

#define _TEST_DATA_ 0

#if _TEST_DATA_
static cJSON* __Test_dial_auth_data()
{
    char* buf = "{\"probeId\":\"1000008\",\"token\":\"C71544C2605146F382EE99FB3E5ADE8C\",\"status\":\"1\"}";
    return cJSON_Parse(buf);
}
static cJSON* __Test_dial_req_data()
{
    char* buf = "{\"dialTaskList\":[{\"dialTaskId\":\"150227934534931\",\"version\":\"1.0\",\"dialTactics\":{\"pingTactics\":[{\"host\":\"10.0.0.210\",\"size\":32,\"port\":18087,\"times\":5,\"timeout\":3,\"businessTag\":\"5\",\"type\":\"tcp\",\"kafkaTopic\":\"pingTopic\"},{\"host\":\"10.0.0.210\",\"size\":32,\"port\":0,\"times\":5,\"timeout\":3,\"businessTag\":\"6\",\"type\":\"icmp\",\"kafkaTopic\":\"pingTopic\"}],\"udpUpTactics\":[],\"httpTactics\":[{\"httpUrlVal\":\"https://hao.qq.com\",\"port\":0,\"timeout\":10,\"businessTag\":\"1\",\"kafkaTopic\":\"httpTopic\"},{\"httpUrlVal\":\"http://www.qq.com\",\"port\":80,\"timeout\":10,\"businessTag\":\"2\",\"kafkaTopic\":\"httpTopic\"}],\"tcpTactics\":[{\"tcpUrlVal\":\"192.168.1.1\",\"port\":45,\"timeout\":2,\"businessTag\":\"3\",\"kafkaTopic\":\"tcpTopic\"},{\"tcpUrlVal\":\"192.168.1.2\",\"port\":46,\"timeout\":2,\"businessTag\":\"4\",\"kafkaTopic\":\"tcpTopic\"}],\"httpDownloadTactics\":[{\"urls\":\"\",\"downTime\":15,\"ignoreTime\":5,\"threads\":4,\"businessTag\":\"11\",\"kafkaTopic\":\"speedTopic\"}],\"tracerouteTactics\":[{\"toAddress\":\"180.97.33.108\",\"size\":32,\"maxNum\":20,\"type\":\"udp\",\"timeout\":3,\"businessTag\":\"7\",\"kafkaTopic\":\"tracerouteTopic\"},{\"toAddress\":\"180.97.33.108\",\"size\":32,\"maxNum\":20,\"type\":\"icmp\",\"timeout\":3,\"businessTag\":\"8\",\"kafkaTopic\":\"tracerouteTopic\"}],\"httpUploadTactics\":[{\"url\":\"\",\"threads\":4,\"upTime\":15,\"ignoreTime\":5,\"businessTag\":\"11\",\"kafkaTopic\":\"speedTopic\"}],\"dnsTactics\":[{\"ip\":\"123.1.2.3\",\"port\":53,\"domainName\":\"www.wangding.com\",\"type\":\"udp\",\"businessTag\":\"9\",\"kafkaTopic\":\"dnsTopic\",\"timeout\":3},{\"ip\":\"123.2.3.5\",\"port\":53,\"domainName\":\"www.wangding2.com\",\"type\":\"tcp\",\"businessTag\":\"10\",\"kafkaTopic\":\"dnsTopic\",\"timeout\":3}]},\"startTime\":1483203661,\"endTime\":1510246861,\"cycleTime\":18000,\"collectTime\":5,\"collectType\":1,\"kafkaUrl\":\"10.0.0.210:9092\"},{\"dialTaskId\":\"1504030622457\",\"version\":\"1.2\",\"dialTactics\":{\"pingTactics\":[],\"udpUpTactics\":[],\"httpTactics\":[{\"httpUrlVal\":\"http://maven.outofmemory.cn/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022003\",\"kafkaTopic\":\"httpTopic\"},{\"httpUrlVal\":\"http://maven.outofmemory.cn/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022003\",\"kafkaTopic\":\"httpTopic\"},{\"httpUrlVal\":\"http://maven.outofmemory.cn/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022003\",\"kafkaTopic\":\"httpTopic\"}],\"tcpTactics\":[],\"httpDownloadTactics\":[],\"tracerouteTactics\":[],\"httpUploadTactics\":[],\"dnsTactics\":[]},\"startTime\":1504159320,\"endTime\":1506060120,\"cycleTime\":60,\"collectTime\":30,\"collectType\":0,\"kafkaUrl\":\"10.0.0.210:9092\"},{\"dialTaskId\":\"1503285489320\",\"version\":\"2.2\",\"dialTactics\":{\"pingTactics\":[],\"udpUpTactics\":[],\"httpTactics\":[{\"httpUrlVal\":\"https://www.baidu.com/\",\"port\":0,\"timeout\":10,\"businessTag\":\"022001,022002\",\"kafkaTopic\":\"httpTopic\"}],\"tcpTactics\":[],\"httpDownloadTactics\":[],\"tracerouteTactics\":[],\"httpUploadTactics\":[],\"dnsTactics\":[]},\"startTime\":1503458160,\"endTime\":1506136560,\"cycleTime\":10,\"collectTime\":5,\"collectType\":0,\"kafkaUrl\":\"10.0.0.210:9092\"}],\"status\":\"1\"}";
    return cJSON_Parse(buf);
}
#endif

int platform_init(Platform_t* platform, char* svr_host, uint16_t tcp_port, uint16_t udp_port, char* scheme)
{
    load_probe_version(PROBE_VERSION_FILE, platform->comm_req.version, sizeof(platform->comm_req.version));
    strcpy(platform->addr.host, svr_host);
    strcpy(platform->addr.scheme, scheme);
    platform->addr.tcp_port = tcp_port;
    platform->addr.udp_port = udp_port;
    TAILQ_INIT(&platform->resp.task.head);

    platform->tpool = thread_pool_new(1);
    thread_pool_set_name(platform->tpool, "platform task");
    thread_pool_set_auto_extern(platform->tpool);

    platform->resp.task.tpool = thread_pool_new(1);
    thread_pool_set_name(platform->resp.task.tpool, "Task");
    thread_pool_set_auto_extern(platform->resp.task.tpool);
    return 1;
}

int platform_auth(Platform_t* platform)
{
    char url[1024] = { 0 };
    char* req = NULL;
    int ret = 0;
    cJSON *resp = NULL, *root = NULL, *json_ptr = NULL;
    struct sockaddr_in addr; //for udp trans

    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port,
        WEB_AUTH_PATH, url);

    if (platform->is_bridge) { //bridge set cmdId 6 for auth
        crq_set_cmdId(&platform->comm_req, 6, 0, 0, NULL);
    }
    root = crq2cjson(&platform->comm_req, root);
    root = devinfo2cjson(&platform->devinfo, root);

#if _TEST_DATA_
    resp = __Test_dial_auth_data();
#else
    if (!platform->is_bridge) {
        req = cJSON_Print(root);
        if (!(resp = http_post(url, req))) {
            log_error("Auth:http post %s error.", url);
            goto out;
        }
    } else {
        cJSON_AddStringToObject(root, "cmdName", "auth");
        req = cJSON_Print(root);
        sockaddr_set(&addr, AF_INET, htons(platform->bridge.port), platform->bridge.host);
        if (!(resp = udp_post(&addr, req, strlen(req)))) {
            log_error("Auth:udp post error.");
            goto out;
        }
    }

#endif

    if (!cJSON_IsObject(resp)) {
        log_error("Auth:JSON Parse failed.");
        goto out;
    }

    if (!check_platform_resp(resp)) {
        goto out;
    }

    cJSON_getvalue(resp, "token", cJSON_String, platform->resp.token);
    cJSON_getvalue(resp, "probeId", cJSON_String, platform->comm_req.probeId);

    ret = 1;
out:
    crq_reset_cmdId(&platform->comm_req);
    cJSON_Delete(root);
    free(req);
    if (resp)
        cJSON_Delete(resp);
    return ret;
}
static cJSON* _dial_req_gen_data(Platform_t* platform, cJSON* root)
{
    int i = 0;
    struct Dial_Task_s* item;
    cJSON *obj, *_root;

    if (!root)
        _root = cJSON_CreateObject();
    else
        _root = root;

    cJSON* tasklist = cJSON_CreateArray();

    cJSON_AddStringToObject(_root, "token", platform->resp.token);

    TAILQ_FOREACH(item, &platform->resp.task.head, entry)
    {
        obj = cJSON_CreateObject();
        cJSON_AddStringToObject(obj, "dialTaskId", item->id);
        cJSON_AddStringToObject(obj, "version", item->version);
        cJSON_AddItemToArray(tasklist, obj);
    }
    cJSON_AddItemToObject(_root, "dialtaskList", tasklist);
    return _root;
}
int platform_dial_req(Platform_t* platform)
{
    char url[1024] = { 0 };
    char* req;
    int arr_len, i = 0, ret = 0;
    cJSON *resp, *root = NULL, *tasklist = NULL, *json_ptr = NULL, *json_obj = NULL;
    Dial_Task_t* dial_task;
    struct sockaddr_in addr;

    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port,
        WEB_DIAL_REQ_PATH, url);

    if (platform->is_bridge) { //bridge set cmdId 7 for req task
        crq_set_cmdId(&platform->comm_req, 7, 0, 0, NULL);
    }
    root = crq2cjson(&platform->comm_req, root);
    root = _dial_req_gen_data(platform, root);

#if _TEST_DATA_
    resp = __Test_dial_req_data();
#else
    if (!platform->is_bridge) {
        req = cJSON_Print(root);
        if (!(resp = http_post(url, req))) {
            log_error("Dial Task Request:http post %s error.", url);
            goto out;
        }
    } else {
        cJSON_AddStringToObject(root, "cmdName", "dialTask");
        req = cJSON_Print(root);
        sockaddr_set(&addr, AF_INET, htons(platform->bridge.port), platform->bridge.host);
        if (!(resp = udp_post(&addr, req, strlen(req)))) {
            log_error("Dial Task:udp post error.");
            goto out;
        }
    }
#endif
    if (!cJSON_IsObject(resp)) {
        log_error("Dial Task Request:response is not Object.");
        goto out;
    }
    if (!check_platform_resp(resp)) {
        goto out;
    }

    tasklist = cJSON_GetObjectItem(resp, "dialTaskList");
    if (!cJSON_IsArray(tasklist)) {
        log_error("Dial Task Request:Dial Task is not Array.");
        goto out;
    }
    // if (cJSON_GetArraySize(tasklist) > 0) {
    //     dial_task_free_all(platform);
    //     TAILQ_INIT(&platform->resp.task_head);
    // }
    arr_len = cJSON_GetArraySize(tasklist);
    platform->resp.task.count = arr_len;

    for (i = 0; i < arr_len; i++) {
        dial_task = calloc(1, sizeof(Dial_Task_t));

        json_obj = cJSON_GetArrayItem(tasklist, i);
        cJSON_getvalue(resp, "dialTaskId", cJSON_String, dial_task->id);
        // json_ptr = cJSON_GetObjectItem(json_obj, "dialTaskId");
        // strcpy(dial_task->id, json_ptr->valuestring);

        cJSON_getvalue(resp, "version", cJSON_String, dial_task->version);
        // json_ptr = cJSON_GetObjectItem(json_obj, "version");
        // strcpy(dial_task->version, json_ptr->valuestring);

        cJSON_getvalue(resp, "startTime", cJSON_Number, (void*)&dial_task->startTime);
        // json_ptr = cJSON_GetObjectItem(json_obj, "startTime");
        // dial_task->startTime = json_ptr->valueint;

        cJSON_getvalue(resp, "endTime", cJSON_Number, (void*)&dial_task->endTime);
        // json_ptr = cJSON_GetObjectItem(json_obj, "endTime");
        // dial_task->endTime = json_ptr->valueint;

        cJSON_getvalue(resp, "cycleTime", cJSON_Number, (void*)&dial_task->cycleTime);
        // json_ptr = cJSON_GetObjectItem(json_obj, "cycleTime");
        // dial_task->cycleTime = json_ptr->valueint;

        cJSON_getvalue(resp, "collectTime", cJSON_Number, (void*)&dial_task->cycleTime);
        // json_ptr = cJSON_GetObjectItem(json_obj, "collectTime");
        // dial_task->collectTime = json_ptr->valueint;

        cJSON_getvalue(resp, "kafkaUrl", cJSON_String, dial_task->kafka_url);
        // json_ptr = cJSON_GetObjectItem(json_obj, "kafkaUrl");
        // strcpy(dial_task->kafka_url, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "dialTactics");
        dial_tactics_parse(&dial_task->tactics, json_ptr);

        TAILQ_INSERT_TAIL(&platform->resp.task.head, dial_task, entry);
    }
    ret = 1;
out:
    crq_reset_cmdId(&platform->comm_req);
    cJSON_Delete(root);
    free(req);
    if (resp)
        cJSON_Delete(resp);
    return ret;
}
#if 0
static void* _each_task_ping_report(void* arg)
{
    void** arg_ptr = arg;
    Platform_t* platform = (Platform_t*)arg_ptr[0];
    ping_tactics_t* ptr = (ping_tactics_t*)arg_ptr[1];
    free(arg);

    char* post_data = NULL, reportTime[32], collectTime[32];
    cJSON *data;

    while (1) {
        pthread_mutex_lock(&ptr->mutex);
        pthread_cond_wait(&ptr->cond, &ptr->mutex);

        data = cJSON_CreateObject();
        cJSON_AddStringToObject(data, "dialTaskId", ptr->taskId);
        cJSON_AddStringToObject(data, "version", ptr->version);
        cJSON_AddStringToObject(data, "indexType", "ping");
        get_timestring(reportTime);
        cJSON_AddStringToObject(data, "reportTime", reportTime);
        get_timestring(collectTime);
        cJSON_AddStringToObject(data, "collectTime", collectTime);
        cJSON_AddItemToObject(data, "pingList", ptr->result);
        post_data = cJSON_Print(data);
        log_debug("post_data:%s", post_data);
        cJSON_Delete(ptr->result);
        free(post_data);
        pthread_mutex_unlock(&ptr->mutex);
    }
    return NULL;
}
static void* _each_task_report(void* arg)
{
    void** arg_ptr = arg;
    cJSON* report_data;
    Platform_t* platform = (Platform_t*)arg_ptr[0];
    struct Dial_Task_s* ptr = (struct Dial_Task_s*)arg_ptr[1];
    free(arg);

    thread_pool_t* tpool = thread_pool_new(1);
    thread_pool_set_auto_extern(tpool);
    thread_pool_set_name(tpool, "task report");

    arg_ptr = calloc(2, sizeof(void*));
    arg_ptr[0] = (void*)platform;
    arg_ptr[1] = (void*)&ptr->tactics.ping;
    strcpy(ptr->tactics.ping.taskId, ptr->id);
    strcpy(ptr->tactics.ping.version, ptr->version);
    thread_pool_add_task(tpool, _each_task_ping_report, (void*)arg_ptr);

    thread_pool_wait(tpool);
    thread_pool_destroy(tpool);

    return NULL;
}
void* platform_report(void* arg)
{
    thread_pool_t* tpool = thread_pool_new(1);
    thread_pool_set_auto_extern(tpool);
    thread_pool_set_name(tpool, "platform report");

    Platform_t* platform = (Platform_t*)arg;
    struct Dial_Task_Head* head = &platform->resp.task_head;
    struct Dial_Task_s* ptr;
    void** thread_arg;
    TAILQ_FOREACH(ptr, head, entry)
    {
        thread_arg = calloc(2, sizeof(void*));
        thread_arg[0] = (void*)platform;
        thread_arg[1] = (void*)ptr;
        thread_pool_add_task(tpool, _each_task_report, (void*)thread_arg);
    }
    thread_pool_wait(tpool);
    thread_pool_destroy(tpool);

    return NULL;
}
#endif
int platform_push_probe_upload_report(Platform_t* platform, int cmd_id, int code)
{
    cJSON *root = NULL, *resp;
    char* comm_req = NULL;
    char url[512];

    crq_set_cmdId(&platform->comm_req, cmd_id, 0, code, NULL);
    root = crq2cjson(&platform->comm_req, root);
    comm_req = cJSON_Print(root);
    cJSON_Delete(root);
    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port, WEB_PUSH_PROBE_UPLOAD_PATH, url);
    if ((resp = http_post(url, comm_req)) == NULL) {
        log_error("HTTP Push probe upload failed.");
        goto out;
    }
    if (resp)
        cJSON_Delete(resp);
out:
    crq_reset_cmdId(&platform->comm_req);
    free(comm_req);
    return 1;
}
int platform_download_probe(Platform_t* platform)
{
    char version[32] = { 0 };
    char url[512], *buffer, *req, down_url[512];
    int len;
    FILE* fp;
    cJSON *root, *resp;
    int ret = 0;

    root = crq2cjson(&platform->comm_req, root);
    req = cJSON_Print(root);
    cJSON_Delete(root);

    http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port, WEB_DOWNLOAD_PROBE_PATH, url);
    if (http_download_get_url(url, req, down_url, version) == 0) {
        log_error("HTTP download get down url failed");
        goto out;
    }

    if (strcmp(version, platform->comm_req.version) == 0) {
        log_info("Local Probe version is same as remote version, no need download.");
        ret = 1;
        goto out;
    }

    save_probe_version(PROBE_VERSION_FILE, version, strlen(version));
    load_probe_version(PROBE_VERSION_FILE, platform->comm_req.version, sizeof(platform->comm_req.version));

    if (http_download_file(down_url, PROBE_TAR_FILE) == 0) {
        log_error("HTTP download Probe failed.");
        goto out;
    }

    if (0 != system("tar xzf " PROBE_TAR_FILE)) {
        if (0 != system("busybox tar xzf " PROBE_TAR_FILE)) {
            log_syserr("Untar " PROBE_TAR_FILE " failed.");
            goto out;
        }
    }
    ret = 1;
out:
    free(req);
    return ret;
}
static void _set_nodeurl(Platform_t* platform)
{
    struct Dial_Task_s* ptr;
    TAILQ_FOREACH(ptr, &platform->resp.task.head, entry)
    {
        http_url(platform->addr.scheme, platform->addr.host, platform->addr.tcp_port,
            WEB_SPEED_NODE_PATH, ptr->nodeUrl);
        strcpy(ptr->probeId, platform->comm_req.probeId);
    }
}
static void _deal_cmd_id(Platform_t* platform, uint32_t cmd_id)
{
    int flag = 0;

    log_trace("Deal cmd id:%d", cmd_id);
    switch (cmd_id) {
    case 1: //pushProbe
        if (!platform_download_probe(platform)) {
            platform_push_probe_upload_report(platform, cmd_id, 0);
        } else {
            platform_push_probe_upload_report(platform, cmd_id, 1);
            log_info("Probe Update successfully, Restart...");
            exit(0);
        }
        break;
    case 2: //pushDialTactics
        dial_task_free_all(&platform->resp.task);
        if (!platform_auth(platform)) {
            log_warn("Platform Auth failed.");
            platform_push_probe_upload_report(platform, cmd_id, 0);
            exit(0);
        }

        if (!platform_dial_req(platform)) {
            log_warn("Platform request dial task failed.");
            platform_push_probe_upload_report(platform, cmd_id, 0);
            exit(0);
        }
        _set_nodeurl(platform);
        thread_pool_add_task(platform->tpool, dial_task_run, (void*)&platform->resp.task.head);
        break;
    case 31: //probeCtrl
    case 32:
        if (cmd_id == 31)
            flag = DIAL_TASK_FLAG_RUN;
        if (cmd_id == 32)
            flag = DIAL_TASK_FLAG_STOP;
        dial_task_all_set_flag(&platform->resp.task, flag);
        platform_push_probe_upload_report(platform, cmd_id, 1);
        break;
    default:
        log_error("Unkown cmd id:%d", cmd_id);
        break;
    }
}
static void* platform_check_lastcmd(void* arg)
{
    Platform_t* platform = (Platform_t*)arg;
    struct sockaddr_un un;
    int fd, cli_fd;
    int ret;
    uint32_t cmd_id;

    fd = socket(AF_UNIX, SOCK_STREAM, 0);
    un.sun_family = AF_UNIX;
    strcpy(un.sun_path, HEART_UNIX_PATH);
    unlink(HEART_UNIX_PATH);

    if (bind(fd, (struct sockaddr*)&un, sizeof(un)) < 0) {
        log_syserr("Platform Check LastCmd bind error.");
        close(fd);
        return NULL;
    }
    if (listen(fd, 10) < 0) {
        log_syserr("Platform Check LastCmd listen error.");
        close(fd);
        return NULL;
    }
    while (1) {
        cmd_id = 0;
        cli_fd = accept(fd, NULL, NULL);
        if (cli_fd > 0) {
            read(cli_fd, &cmd_id, sizeof(cmd_id));
            _deal_cmd_id(platform, cmd_id);
            close(cli_fd);
        }
    }
}
int platform_task_run(Platform_t* platform)
{
    thread_pool_t* tpool = platform->tpool;

    _set_nodeurl(platform);

    thread_pool_add_task(tpool, platform_check_lastcmd, (void*)platform);
    thread_pool_add_task(tpool, heart_bit_loop, (void*)&platform->heartinfo);
    thread_pool_add_task(tpool, dial_task_run, (void*)&platform->resp.task.head);
    thread_pool_wait(tpool);
    thread_pool_destroy(tpool);
    return 1;
}